package com.atguigu.dataStreamApi.transformation;

import com.alibaba.fastjson.JSON;
import com.atguigu.pojo.Event;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;
import util.SourceUtil;

public class Flink01_BaseOperator {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<Event> ds
                = env.fromSource(SourceUtil.getSource(), WatermarkStrategy.noWatermarks(), "dataGenSource");


        ds.print("INPUT");

        //map:将Event数据转换成Json格式的字符串输出
        SingleOutputStreamOperator<String> mapds = ds.map(
                JSON:: toJSONString
        );
        SingleOutputStreamOperator<Event> filterds = ds.filter(
                value -> !("Tom".equals(value.getUser()) || "Jerry".equals((value.getUser()))));

        //flatmap：将Event中user为"Tom"和"Jerry"的数据过滤掉，同时将Event的三个属性拆分输出

        SingleOutputStreamOperator<String> flatMapds = ds.flatMap(
                new FlatMapFunction<Event, String>() {
                    @Override
                    public void flatMap(Event value, Collector<String> out) throws Exception {
                        if (!("Tom".equals(value.getUser()) || "Jerry".equals(value.getUser()))){

                            //user
                            out.collect(value.getUser());

                            out.collect(value.getUrl());

                            out.collect(String.valueOf(value.getTs()));

                        }
                    }
                }
        );
        flatMapds.print("FLATMAP");

        try {
            env.execute();
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    }

    }

